package com.jw.debezuim.config.debezium;

import com.alibaba.fastjson.JSONObject;
import io.debezium.connector.SnapshotRecord;
import io.debezium.connector.mysql.SourceInfo;
import io.debezium.data.Envelope;
import io.debezium.embedded.StopConnectorException;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.StopEngineException;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.json.JsonConverter;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;

/**
 * @auther wangjianying
 * @date 2023/04/07 15:44
 */
@Slf4j
public class EventChangeConsumer  implements DebeziumEngine.ChangeConsumer<SourceRecord> {
    @Override
    public void handleBatch(List<SourceRecord> records, DebeziumEngine.RecordCommitter<SourceRecord> committer) throws InterruptedException {
        log.info("批处理数量=>{}", records.size());
        for (SourceRecord record : records) {
            try {
                doHandleSimple(record);
            }catch (Throwable ex) {
                log.error("处理记录异常捕获", ex);
            }finally {
                committer.markProcessed(record);
            }
        }
        committer.markBatchFinished();
    }

    @Override
    public boolean supportsTombstoneEvents() {
        return DebeziumEngine.ChangeConsumer.super.supportsTombstoneEvents();
    }

    private void doHandleSimple(SourceRecord record) {
        String topic = record.topic();
        switch (topic) {
            case "zt3000-debezuim-mysql-connector.test_binlog.tb_metadata_plat_cat":
            case "zt3000-debezuim-mysql-connector.jw_test.product_view":
                Struct struct = (Struct) record.value();
                Struct key = (Struct)record.key();
                log.info("record=>{}", record);
                log.info("topic=>{}", topic);
                Struct source = struct.getStruct(Envelope.FieldName.SOURCE);
                String db = source.getString("db");
                String table = source.getString("table");
                Struct before = struct.getStruct(Envelope.FieldName.BEFORE);
                Struct after = struct.getStruct(Envelope.FieldName.AFTER);
                Envelope.Operation operation = Envelope.Operation.forCode(struct.getString(Envelope.FieldName.OPERATION));
                log.info("db=>{}", db);
                log.info("table=>{}", table);
                log.info("before=>{}", toJson(before));
                log.info("after=>{}", toJson(after));
                log.info("operation=>{}", operation);
                break;
            default:
                log.info("do handle SourceRecord ignored");
                break;
        }
    }

    private JSONObject toJson(Struct struct) {
        if (struct == null) {
            return null;
        }
        JSONObject jsonObject = new JSONObject();
        List<Field> fields = struct.schema().fields();
        for (int i = 0; i < fields.size(); i++) {
            Field field = fields.get(i);
            String name = field.name();
            jsonObject.put(name, struct.get(name));
        }
        return jsonObject;

    }
}
